from typing import Dict, List, Any
import asyncio
from datetime import datetime
import json

class MessageQueueManager:
    _instance = None
    
    def __new__(cls):
        if cls._instance is None:
            cls._instance = super().__new__(cls)
            # 初始化队列
            cls._instance.message_queue = asyncio.Queue()
            # 存储订阅者
            cls._instance.subscribers = {} # : Dict[str, asyncio.Queue]
            # 存储历史消息
            cls._instance.history = [] # : List[Dict]
            # 最大历史消息数量
            cls._instance.max_history = 100
        return cls._instance
    
    async def publish(self, message: Any, source: str = "unknown"):
        """
        发布消息到队列
        :param message: 消息内容
        :param source: 消息来源（http/socket/websocket等）
        """
        message_data = {
            "content": message,
            "source": source,
            "timestamp": datetime.now().isoformat()
        }
        
        # 添加到历史记录
        self.history.append(message_data)
        if len(self.history) > self.max_history:
            self.history.pop(0)
        
        # 发送到队列
        await self.message_queue.put(message_data)
        
        # 广播给所有订阅者
        for subscriber_queue in self.subscribers.values():
            await subscriber_queue.put(message_data)
    
    async def subscribe(self, subscriber_id: str) -> asyncio.Queue:
        """
        订阅消息
        :param subscriber_id: 订阅者ID
        :return: 订阅者的消息队列
        """
        if subscriber_id not in self.subscribers:
            self.subscribers[subscriber_id] = asyncio.Queue()
        return self.subscribers[subscriber_id]
    
    async def unsubscribe(self, subscriber_id: str):
        """取消订阅"""
        if subscriber_id in self.subscribers:
            del self.subscribers[subscriber_id]
    
    def get_history(self, limit: int = None) -> List[Dict]:
        """获取历史消息"""
        if limit is None:
            return self.history
        return self.history[-limit:]
    
    async def clear_history(self):
        """清空历史消息"""
        self.history.clear() 